﻿using Confluent.Kafka;
using Micro.Extensions.Utilities;
using Microsoft.Extensions.Options;
using System;

namespace Micro.Extensions.Log.Kafka
{
    public class KafkaLoggerWriter : ILoggerWriter, IDisposable
    {
        private readonly IKafkaClient _kafkaClient;

        private readonly IOptions<KafkaClientConfig> _options;
        private readonly IFileDiskWriter _diskWriter;

        public KafkaLoggerWriter(IKafkaClient kafkaClient, IOptions<KafkaClientConfig> options, IFileDiskWriter diskWriter)
        {
            _kafkaClient = kafkaClient;
            _options = options;
            _diskWriter = diskWriter;
        }
        //异步写入回调
        private void AsyncHandler(DeliveryReport<Null, string> deliveryReport)
        {
            if (deliveryReport.Error.IsError)
            {
                _diskWriter.Write(new HardDiskInfo
                {
                    FileName = LogConst.ErrorLog, 
                    Message = new BaseMessage {Message = deliveryReport.Error.ToJson()}
                });
            }
        }

        /// <summary>
        /// 日志写入kafka
        /// </summary>
        /// <param name="baseMessage"></param>
        /// <returns></returns>
        public void Write(BaseMessage baseMessage)
        {
            var producer = _kafkaClient.Producer();
            //异步写入
             producer.Produce(
                _options.Value.Topic,
                new Message<Null, string>() { Value = baseMessage.ToJson() }, AsyncHandler);
        }

        public void Dispose()
        {
            _kafkaClient?.Dispose();
        }
    }
}
